/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.database.entity.DashboardHistoryDto;
import spring.cloud.tasks.api.database.entity.StatisticsDto;
import spring.cloud.tasks.api.database.service.StatisticsDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.*;
import spring.cloud.tasks.api.service.helper.DashboardConstants;
import spring.cloud.tasks.api.service.impl.DashboardServiceImpl;
import spring.cloud.tasks.api.service.impl.statistics.analyzer.*;
import spring.cloud.tasks.api.utils.ConsoleThreadFactory;
import spring.cloud.tasks.api.utils.StatisticsTableKeyConstant;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.*;

/**
 * @author timmy.hu
 */
public class StatisticsRefreshServiceImpl implements StatisticsRefreshService {

    private static final Logger log = LoggerFactory.getLogger(StatisticsRefreshServiceImpl.class);

    private static final int STAT_THREAD_NUM = 20;
    private final Map<String/** domainName_jobName_shardingItemStr **/
            , AbnormalShardingState /** abnormal sharding state */> abnormalShardingStateCache = new ConcurrentHashMap<>();
    private Timer refreshStatisticsTimer;
    private Timer cleanAbnormalShardingCacheTimer;
    @Resource
    private StatisticsDtoService statisticsService;

    @Resource
    private StatisticsPersistence statisticsPersistence;


    @Resource
    private ZooKeeperService zooKeeperService;

    @Resource
    private TaskService taskService;

    @Resource
    private ReportAlarmService reportAlarmService;

    private ExecutorService statExecutorService;

    @Resource
    private DashboardService dashboardService;

    @PostConstruct
    public void init() {
        initStatExecutorService();
        startRefreshStatisticsTimer();
        startCleanAbnormalShardingCacheTimer();
    }

    @PreDestroy
    public void destroy() {
        if (statExecutorService != null) {
            statExecutorService.shutdownNow();
        }
        if (refreshStatisticsTimer != null) {
            refreshStatisticsTimer.cancel();
        }
        if (cleanAbnormalShardingCacheTimer != null) {
            cleanAbnormalShardingCacheTimer.cancel();
        }
    }

    private void initStatExecutorService() {
        if (statExecutorService != null) {
            statExecutorService.shutdownNow();
        }
        ThreadPoolExecutor tp = new ThreadPoolExecutor(STAT_THREAD_NUM, STAT_THREAD_NUM,
                DashboardConstants.REFRESH_INTERVAL_IN_MINUTE + 1, TimeUnit.MINUTES,
                new LinkedBlockingQueue<Runnable>(), new ConsoleThreadFactory("dashboard-statistics-thread", true));
        tp.allowCoreThreadTimeOut(true);
        statExecutorService = tp;
    }

    private void startRefreshStatisticsTimer() {
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                try {
                    log.info("start refresh statistics on timer");
                    Date start = new Date();
                    List<ZooKeeperCluster> zooKeeperClusterList = zooKeeperService.getZooKeeperClusterList();
                    if (zooKeeperClusterList != null) {
                        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterList) {
                            if (zooKeeperCluster.isOffline()) {
                                log.info("zkcluster:{} is offline, skip statistics refresh.",
                                        zooKeeperCluster.getZooKeeperClusterKey());
                                continue;
                            }

                            refreshStatistics2DB(zooKeeperCluster);
                        }
                    }
                    log.info("end refresh statistics on timer which takes {}ms",
                            new Date().getTime() - start.getTime());
                } catch (Throwable t) {
                    log.error(t.getMessage(), t);
                }
            }
        };
        refreshStatisticsTimer = new Timer("refresh-statistics-to-db-timer", true);
        refreshStatisticsTimer
                .scheduleAtFixedRate(timerTask, 1000L * 15, 1000L * 60 * DashboardConstants.REFRESH_INTERVAL_IN_MINUTE);
    }

    private void startCleanAbnormalShardingCacheTimer() {
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                try {
                    for (Entry<String, AbnormalShardingState> entrySet : abnormalShardingStateCache.entrySet()) {
                        AbnormalShardingState shardingState = entrySet.getValue();
                        if (shardingState.getAlertTime() + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS * 2 < System
                                .currentTimeMillis()) {
                            abnormalShardingStateCache.remove(entrySet.getKey());
                            log.info("Clean AbnormalShardingCache with key: {}, alertTime: {}, zkNodeCVersion: {}: ",
                                    entrySet.getKey(), shardingState.getAlertTime(), shardingState.getZkNodeCVersion());
                        }
                    }
                } catch (Throwable t) {
                    log.error("Clean AbnormalShardingCache error", t);
                }
            }
        };
        cleanAbnormalShardingCacheTimer = new Timer("clean-abnormalShardingCache-timer", true);
        cleanAbnormalShardingCacheTimer
                .scheduleAtFixedRate(timerTask, 0, DashboardConstants.ALLOW_DELAY_MILLIONSECONDS);
    }

    @Override
    public void refresh(String zkClusterKey, boolean isForce) throws RuntimeException {
        if (isForce) {
            refresh(zkClusterKey);
            return;
        }

        refresh(zkClusterKey);
    }

    private void refresh(String zkClusterKey) throws RuntimeException {
        ZooKeeperCluster zkCluster = zooKeeperService.getZooKeeperCluster(zkClusterKey);
        if (zkCluster != null) {
            refreshStatistics2DB(zkCluster);
        } else {
            throw new RuntimeException("zk cluster not found by zkClusterKey:" + zkClusterKey);
        }
    }

    protected void postRefreshStatistics2DB(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster) {
        statisticsModel.getOutdatedNoRunningJobAnalyzer().reportAlarmOutdatedNoRunningJobs();
        calculateDashboardHistory(statisticsModel, zkCluster);
    }

    private void calculateDashboardHistory(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster) {
        List<DashboardHistoryDto> dashboardHistories = new ArrayList<>();
        Date currentDate = new Date();
        calculateRunTimes(statisticsModel, zkCluster, dashboardHistories, currentDate);
        calculateContainerCount(statisticsModel, zkCluster, dashboardHistories, currentDate);
        calculateJobCount(statisticsModel, zkCluster, dashboardHistories, currentDate);
        calculateDomainCount(statisticsModel, zkCluster, dashboardHistories, currentDate);
        dashboardService.batchSaveDashboardHistory(dashboardHistories);
    }

    private void calculateDomainCount(StatisticsModel statisticsModel, ZooKeeperCluster zooKeeperCluster,
                                      List<DashboardHistoryDto> dashboardHistories, Date currentDate) {
        Map<String, Integer> domainContent = new HashMap<>(1);
        int domainCount = zooKeeperService.domainCount(zooKeeperCluster.getZooKeeperClusterKey());
        domainContent.put("domainCount", domainCount);
        DashboardHistoryDto domainHistory = new DashboardHistoryDto(zooKeeperCluster.getZooKeeperClusterKey(),
                DashboardServiceImpl.DashboardType.DOMAIN.name(),
                DashboardServiceImpl.DashboardTopic.DOMAIN_COUNT.name(), JSON.toJSONString(domainContent), currentDate);
        dashboardHistories.add(domainHistory);
    }

    private void calculateJobCount(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster,
                                   List<DashboardHistoryDto> dashboardHistories, Date currentDate) {
        Map<String, Integer> jobContent = new HashMap<>(1);
        int jobCount = statisticsModel.getJobStatisticsAnalyzer().getJobList().size();
        jobContent.put("jobCount", jobCount);
        DashboardHistoryDto jobHistory = new DashboardHistoryDto(zkCluster.getZooKeeperClusterKey(),
                DashboardServiceImpl.DashboardType.JOB.name(), DashboardServiceImpl.DashboardTopic.JOB_COUNT.name(),
                JSON.toJSONString(jobContent), currentDate);
        dashboardHistories.add(jobHistory);
    }

    private void calculateContainerCount(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster,
                                         List<DashboardHistoryDto> dashboardHistories, Date currentDate) {
        Map<String, Integer> executorContent = new HashMap<>(2);
        int inDocker = statisticsModel.getExecutorInfoAnalyzer().getExeInDocker();
        int notInDocker = statisticsModel.getExecutorInfoAnalyzer().getExeNotInDocker();
        executorContent.put("dockerCount", inDocker);
        executorContent.put("otherCount", notInDocker);
        DashboardHistoryDto executorHistory = new DashboardHistoryDto(zkCluster.getZooKeeperClusterKey(),
                DashboardServiceImpl.DashboardType.EXECUTOR.name(),
                DashboardServiceImpl.DashboardTopic.EXECUTOR_COUNT.name(), JSON.toJSONString(executorContent),
                currentDate);
        dashboardHistories.add(executorHistory);
    }

    private void calculateRunTimes(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster,
                                   List<DashboardHistoryDto> dashboardHistories, Date currentDate) {
        long successCount = statisticsModel.getZkClusterDailyCountAnalyzer().getTotalCount();
        long failCount = statisticsModel.getZkClusterDailyCountAnalyzer().getErrorCount();
        Map<String, Long> content = new HashMap<>(2);
        content.put("count", successCount);
        content.put("failCount", failCount);
        DashboardHistoryDto allDomainHistory = new DashboardHistoryDto(zkCluster.getZooKeeperClusterKey(),
                DashboardServiceImpl.DashboardType.DOMAIN.name(),
                DashboardServiceImpl.DashboardTopic.DOMAIN_OVERALL_COUNT.name(), JSON.toJSONString(content),
                currentDate);
        dashboardHistories.add(allDomainHistory);
    }


    private void refreshStatistics2DB(ZooKeeperCluster zooKeeperCluster) {
        log.info("start refresh statistics by zkClusterKey:{}", zooKeeperCluster.getZooKeeperClusterKey());
        Date start = new Date();
        StatisticsModel statisticsModel = initStatisticsModel();
        List<Callable<Boolean>> callableList = getStatCallableList(zooKeeperCluster, statisticsModel);
        try {
            if (callableList != null && !callableList.isEmpty()) {
                statExecutorService.invokeAll(callableList);
            }
            statisticsPersistence.persist(statisticsModel, zooKeeperCluster);
            postRefreshStatistics2DB(statisticsModel, zooKeeperCluster);
        } catch (InterruptedException e) {
            log.warn("the refreshStatistics2DB thread is interrupted", e);
            Thread.currentThread().interrupt();
        }
        log.info("end refresh statistics by zkClusterKey:{}, takes {}", zooKeeperCluster.getZooKeeperClusterKey(),
                new Date().getTime() - start.getTime());
    }

    private List<Callable<Boolean>> getStatCallableList(final ZooKeeperCluster zooKeeperCluster,
                                                        final StatisticsModel statisticsModel) {
        List<Callable<Boolean>> callableList = Lists.newArrayList();
        for (final NameSpaceConfiguration nameSpaceConfiguration : zooKeeperCluster.getNameSpaceConfigurationList()) {
            // 过滤非当前zk连接
            if (!zooKeeperCluster.getZooKeeperAddress().equals(nameSpaceConfiguration.getZooKeeperAddressList())) {
                continue;
            }
            Callable<Boolean> callable = new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return analyzeStatistics(statisticsModel, zooKeeperCluster, nameSpaceConfiguration);
                }
            };
            callableList.add(callable);
        }
        return callableList;
    }

    private boolean analyzeStatistics(StatisticsModel statisticsModel, ZooKeeperCluster zooKeeperCluster,
                                      NameSpaceConfiguration nameSpaceConfiguration) {
        String nameSpace = nameSpaceConfiguration.getNameSpace();
        try {
            DomainStatistics domain = statisticsModel.getDomainStatisticsAnalyzer().initDomain(zooKeeperCluster, nameSpaceConfiguration);
            ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
            List<AbnormalJob> oldAbnormalJobs = getOldAbnormalJobs(zooKeeperCluster);
            List<Timeout4AlarmJob> oldTimeout4AlarmJobs = getOldTimeout4AlarmJobs(zooKeeperCluster);

            statisticsModel.analyzeExecutor(zooKeeper, nameSpaceConfiguration);
            List<String> jobs = taskService.getUnSystemJobNames(nameSpaceConfiguration.getNameSpace());
            for (String job : jobs) {
                if (!zooKeeper.checkExists(TaskNodePath.getConfigNodePath(job))) {
                    continue;
                }

                try {
                    JobStatistics jobStatistics = statisticsModel.analyzeJobStatistics(zooKeeper, job, nameSpaceConfiguration);
                    String jobDegree = String.valueOf(jobStatistics.getJobDegree());
                    statisticsModel.analyzeShardingCount(zooKeeper, domain);
                    {// 非本地作业才参与判断
                        statisticsModel.analyzeOutdatedNoRunningJob(zooKeeper, oldAbnormalJobs, job, jobDegree, nameSpaceConfiguration);
                    }
                    statisticsModel.analyzeTimeout4AlarmJob(zooKeeper, oldTimeout4AlarmJobs, job, jobDegree, nameSpaceConfiguration);
                    statisticsModel.analyzeUnableFailoverJob(zooKeeper, job, jobDegree, nameSpaceConfiguration);

                } catch (Exception e) {
                    log.info(String.format("analyzeStatistics namespace(%s) jobName(%s) error", nameSpace, job), e);
                }
            }
            statisticsModel.analyzeProcessCount(domain, jobs, nameSpaceConfiguration);
        } catch (Exception e) {
            log.info(String.format("analyzeStatistics namespace(%s) error", nameSpace), e);
            return false;
        }
        return true;
    }

    private List<AbnormalJob> getOldAbnormalJobs(ZooKeeperCluster zooKeeperCluster) {
        StatisticsDto saturnStatistics = statisticsService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.UNNORMAL_JOB, zooKeeperCluster.getZooKeeperAddress());
        List<AbnormalJob> oldAbnormalJobs = new ArrayList<>();
        if (saturnStatistics != null) {
            String result = saturnStatistics.getResult();
            if (StringUtils.isNotBlank(result)) {
                oldAbnormalJobs = JSON.parseArray(result, AbnormalJob.class);
            }
        }
        return oldAbnormalJobs;
    }

    private List<Timeout4AlarmJob> getOldTimeout4AlarmJobs(ZooKeeperCluster zooKeeperCluster) {
        StatisticsDto saturnStatistics = statisticsService
                .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TIMEOUT_4_ALARM_JOB, zooKeeperCluster.getZooKeeperAddress());
        List<Timeout4AlarmJob> oldTimeout4AlarmJobs = new ArrayList<>();
        if (saturnStatistics != null) {
            String result = saturnStatistics.getResult();
            if (StringUtils.isNotBlank(result)) {
                oldTimeout4AlarmJobs = JSON.parseArray(result, Timeout4AlarmJob.class);
            }
        }
        return oldTimeout4AlarmJobs;
    }


    protected StatisticsModel initStatisticsModel() {
        StatisticsModel statisticsModel = new StatisticsModel();
        ExecutorInfoAnalyzer executorInfoAnalyzer = new ExecutorInfoAnalyzer();
        statisticsModel.setExecutorInfoAnalyzer(executorInfoAnalyzer);

        OutdatedNoRunningJobAnalyzer outdatedNoRunningJobAnalyzer = new OutdatedNoRunningJobAnalyzer();
        outdatedNoRunningJobAnalyzer.setAbnormalShardingStateCache(abnormalShardingStateCache);
        outdatedNoRunningJobAnalyzer.setReportAlarmService(reportAlarmService);
        outdatedNoRunningJobAnalyzer.setTaskService(taskService);
        statisticsModel.setOutdatedNoRunningJobAnalyzer(outdatedNoRunningJobAnalyzer);

        UnableFailoverJobAnalyzer unableFailoverJobAnalyzer = new UnableFailoverJobAnalyzer();
        unableFailoverJobAnalyzer.setExecutorProvidedList(taskService);
        statisticsModel.setUnableFailoverJobAnalyzer(unableFailoverJobAnalyzer);

        Timeout4AlarmJobAnalyzer timeout4AlarmJobAnalyzer = new Timeout4AlarmJobAnalyzer();
        timeout4AlarmJobAnalyzer.setReportAlarmService(reportAlarmService);
        statisticsModel.setTimeout4AlarmJobAnalyzer(timeout4AlarmJobAnalyzer);


        JobStatisticsAnalyzer jobStatisticsAnalyzer = new JobStatisticsAnalyzer();
        statisticsModel.setJobStatisticsAnalyzer(jobStatisticsAnalyzer);

        DomainStatisticsAnalyzer domainStatisticsAnalyzer = new DomainStatisticsAnalyzer();
        statisticsModel.setDomainStatisticsAnalyzer(domainStatisticsAnalyzer);

        ZkClusterDailyCountAnalyzer zkClusterDailyCountAnalyzer = new ZkClusterDailyCountAnalyzer();
        statisticsModel.setZkClusterDailyCountAnalyzer(zkClusterDailyCountAnalyzer);
        return statisticsModel;
    }

}